/*
 * ===================================================================
 *  
 *      Filename:  log_client_protocol_tcp.c
 *
 *   Description:  This file is used to send the packet to the remote
 *                 log server with TCP.
 *
 *       Version:  1.0
 *        Create:  2010/07/13
 *
 *        Author:  Keven Zhu     kevenzhu@tencent.com
 *     Copyright:  Reserved by Tencent.com Co.ltd
 *          
 * ====================================================================
 */
#include <stdio.h>
#include <stdlib.h>
#include <errno.h>
#include <string.h>
#include <netdb.h>
#include <pthread.h>
#include <sys/types.h>
#include <sys/socket.h>
#include <netinet/in.h>
#include <sys/epoll.h>
#include <unistd.h>

#include "lib/util.h"
#include "log_client_api.h"

#define MAX_TCP_BACKLOG        2048
#define MAX_EVENT_NUM          2048
#define MAX_EPOLL_SIZE         2048
int epfd = INV_HANDLER;
pthread_t     epoll_pid;

int global_tcp_sock = INV_HANDLER;

/*main routine for sending the packet in tcp*/
int send_log_via_tcp(char* msg)
{
    int iret = 0;
    int ilen = *((short*)msg);
    if(global_tcp_sock == INV_HANDLER)
    {
        /*send the log to local tmp buffer*/
        log_doctor(doctor_level,"the tcp socket was not valid\n");
        goto ERR_SND_TCP;
    }
    while(ilen)
    {
        iret = send(global_tcp_sock,msg,ilen,0);
        if(-1 == iret)
        {
            perror("error to send the packet with tcp");
            goto ERR_SND_TCP;
        }    
        else
        {
            /*shift the point for the next packet*/
            msg  = msg  + iret;
            ilen = ilen - iret;
        }
    }
    return NOERR;
ERR_SND_TCP:
    send_log_to_backup(msg);
    return INVIO;
}

/*flush all the backup file to the srv*/
int flush_backup_to_srv_with_tcp()
{
    if(global_tcp_sock != INV_HANDLER)
        flush_backup_to_srv(global_tcp_sock);
    return NOERR;
}
/*reinitialize the tcp socket*/
int reinit_protocol_tcp(unsigned int srv_addr)
{
    int tmp_sock = 0,once_sock = 0;
    struct sockaddr_in serv_addr;
    struct timeval timeo;
    struct epoll_event ev;
    socklen_t len = sizeof(timeo);

    if(srv_addr)
    {
        tmp_sock = socket(AF_INET,SOCK_STREAM,0);
        if(tmp_sock == -1)
        {
            perror("error to create the socket");
            goto INIT_ERR;
        }
        /*set the timeout value: 2s */
        memset(&timeo,0,sizeof(timeo));
        timeo.tv_sec  = 2;
        if (setsockopt(tmp_sock, SOL_SOCKET, SO_SNDTIMEO, &timeo, len) == -1)
        {
            perror("setsockopt");
        }        
        if (setsockopt(tmp_sock, SOL_SOCKET, SO_RCVTIMEO, &timeo, len) == -1)
        {
            perror("setsockopt");
        }        
        /*fill the address field*/
        serv_addr.sin_family = AF_INET;
        serv_addr.sin_port   = htons(HPLOG_SRV_PORT);
        serv_addr.sin_addr.s_addr = htonl(srv_addr);
        bzero(&(serv_addr.sin_zero),8);
        /*connect the socket to the srv*/
        if(connect(tmp_sock, (struct sockaddr*)&serv_addr,sizeof(struct sockaddr))== -1)
        {
            close(tmp_sock);
            perror("error to call the tcp connect");
            goto INIT_ERR;
        } 
        /*add to the epoll*/
        ev.data.fd = tmp_sock;
        ev.events  = EPOLLIN ;
        //ev.events  = EPOLLIN |EPOLLLT;
        if(-1 == epoll_ctl(epfd,EPOLL_CTL_ADD,tmp_sock,&ev))
        {
            perror("error to add the new client to the epoll");
        }
        /*replace the global socket*/
        once_sock = global_tcp_sock;
        global_tcp_sock = tmp_sock;
        if(once_sock != INV_HANDLER)
            close(once_sock);
        return NOERR;
    }
    else
    {
        /*didn't fine the srv addr*/
        goto INIT_ERR;
    }
INIT_ERR:
    if(global_tcp_sock != INV_HANDLER)
        close(global_tcp_sock);
    global_tcp_sock = INV_HANDLER;   
    return INVIO;
}

void* epoll_routine(void *argv)
{
    struct epoll_event events[MAX_EVENT_NUM];
    int nfds,i;
    char buff[MAX_LOG_MSG_SIZE];

    /*create the epoll firstly*/
    if(-1 == (epfd = epoll_create(MAX_EPOLL_SIZE)))
    {
        perror("error to create the epoll");
        return NULL;
    }
    while(1)
    {
        nfds = epoll_wait(epfd,events,MAX_EVENT_NUM,3000);
        if(nfds>0)
        {
            /*socket available*/
        }
        else if(!nfds)
        {
            continue;
        }
        else
        {
            continue;
        }
        for(i=0;i<nfds;i++)
        {
            if(recv(events[i].data.fd,buff,MAX_LOG_MSG_SIZE,0))
            {
                continue;
            }
            else
            {
                log_doctor(doctor_level,"error on the socket\n");
                /*set the global tcp to INV_HANDLER*/
                global_tcp_sock = INV_HANDLER;
                close(events[i].data.fd);
            }
        }
    }

}

int init_protocol_tcp()
{
    unsigned int srv_addr = 0;
    if(pthread_create(&epoll_pid,NULL,epoll_routine,NULL))
    {
        perror("error to create the epoll thread");
        return INVIO;
    }    
    /*flush the srv list*/
    srv_addr = get_destination();
    reinit_protocol_tcp(srv_addr);
    return NOERR;
}

int finit_protocol_tcp()
{
    pthread_cancel(epoll_pid);
    if(global_tcp_sock != INV_HANDLER)
        close(global_tcp_sock);
    return NOERR;
}
